package com.gvtv.main.hive;

import java.util.LinkedList;
import java.util.List;
import java.util.regex.MatchResult;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.hive.service.HiveClient;
import org.apache.hadoop.hive.service.HiveServerException;
import org.apache.thrift.TException;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TProtocol;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;

import com.gvtv.main.util.PropertiesHelper;

public class HiveExecute {

	private String host;
	private int port;
	private static Pattern pa = Pattern.compile("//(.+?)[/|?]", Pattern.CASE_INSENSITIVE);
	private HiveClient hiveClient;
	private TTransport transport;
	
	public HiveExecute() {
		setHostPort();
	}

	public HiveExecute(String host, int port) {
		this.host = host;
		this.port = port;
	}

	private void setHostPort() {
		String url = PropertiesHelper.getString("hive.url");
		Matcher ma = pa.matcher(url);
		String hostPort = null;
		if (ma.find()) {
			MatchResult mr = ma.toMatchResult();
			hostPort = mr.group(1);
		}
		if (StringUtils.isNotBlank(hostPort)) {
			String[] ss = hostPort.split(":");
			host = ss[0];
			if (ss.length > 1 && StringUtils.isNumeric(ss[1])) {
				port = Integer.valueOf(ss[1]);
			} else {
				port = 10000;
			}
		}
	}

	private void connect() throws TException {
		transport = new TSocket(host, port);
		TProtocol protocol = new TBinaryProtocol(transport);
		hiveClient = new HiveClient(protocol);
		try {
			transport.open();
			hiveClient.execute("set hive.fetch.output.serde = org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe");
		} catch (TException e) {
			throw e;
		}
	}
	/**
	 * 执行非查询语句
	 * @param sql
	 * @throws TException
	 */
	public void execute(String sql) throws TException{
		try {
			if(hiveClient==null){
				connect();
			}
			hiveClient.execute(sql);
		} catch (HiveServerException e) {
			throw e;
		} catch (TException e) {
			throw e;
		}
	}
	/**
	 * 执行查询语句，查询结果集比较小使用，否则容易导致内存溢出
	 * @param sql
	 * @return
	 * @throws TException
	 */
	public List<String[]> query(String sql) throws TException{
		try {
			if(hiveClient==null){
				connect();
			}
			hiveClient.execute(sql);
			List<String> list = hiveClient.fetchAll();
			if(list==null){
				return null;
			}
			List<String[]> data = new LinkedList<String[]>();
			for (String r : list) {
				data.add(r.split("\001"));
			}
			return data;
		} catch (HiveServerException e) {
			throw e;
		} catch (TException e) {
			throw e;
		}
	}
	public void close(){
		try {
			if(hiveClient!=null){
				hiveClient.clean();
			}
			if(transport!=null){
				transport.close();
			}
		} catch (TException e) {
			e.printStackTrace();
		}
	}
	
	public static void main(String[] args) {
		HiveExecute hiveExecute = new HiveExecute();
		try {
			List<String[]> data = hiveExecute.query("select * from default.user_news limit 2");
			for (String[] ss : data) {
				System.out.println(ss[0]+"  "+ss[1]+"  "+ss[3]);
			}
			
		} catch (TException e) {
			e.printStackTrace();
		} finally{
			hiveExecute.close();
		}
	}
}
